#pragma once
#include "../source/message.hpp"
#include "../source/net.hpp"
#include "../source/dispacher.hpp"
#include "Rpc_router.hpp"
#include <unordered_set>

namespace ns_server_topic
{
    class TopicManager
    {
    public:
        using ptr = std::shared_ptr<TopicManager>;
        TopicManager(){};
        void OnTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            TopicOptype otype = msg->GetOtype();
            bool ret = true;
            switch (otype)
            {
            case TopicOptype::TOPIC_CREATE: TopicCreate(conn,msg); break;
            case TopicOptype::TOPIC_REMOVE: TopicRemove(conn,msg); break;
            case TopicOptype::TOPIC_SUBSCRIBE: ret = TopicSubscribe(conn,msg); break;
            case TopicOptype::TOPIC_CANCEL: TopicCancel(conn,msg); break;
            case TopicOptype::TOPIC_PUBLISH: ret = TopicPublish(conn,msg); break;
            default: ErrorResponse(conn,msg,Rcode::RCODE_INVALID_OPTYPE);
            }
            if(ret) return TopicOKResponse(conn,msg);
            else return ErrorResponse(conn,msg,Rcode::RCODE_NOT_FOUND_TOPIC);
        }
        // 一个订阅者下线，删除对应主题内部管理的关于它的连接。 订阅者内部订阅过的所有主题也要全部释放
        void OnShutDown(const BaseConnection::ptr &conn)
        {
            //1.判断是否是订阅者，不是则返回
            std::vector<Topic::ptr> topics;
            Subscriber::ptr subs;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _hashconns.find(conn);
                if(it == _hashconns.end())
                    return;
                //2.获取到受影响的topic对象，订阅者
                subs = it->second;
                for(auto& topic_name : subs->_topics)
                {
                    auto topic_it = _hashtopics.find(topic_name);
                    if(topic_it == _hashtopics.end()) continue;
                    topics.push_back(topic_it->second);
                }
                //3.删除_hashconns中的映射
                _hashconns.erase(it);
            }
            //4.删除每个主题中关于它的连接
            for(auto& topic : topics)
            {
                topic->RemoveSubscriber(subs);
            }
        }

    private:
        void TopicOKResponse(const BaseConnection::ptr& conn,const TopicRequest::ptr& msg)
        {
            auto msg_rsp = MessageFactory::Createmsg<TopicResponse>();
            msg_rsp->Setid(msg->Getid());
            msg_rsp->SetMtype(Mtype::TOPIC_RES);
            msg_rsp->SetRcode(Rcode::RCODE_OK);
            conn->Send(msg_rsp);
        }
        void ErrorResponse(const BaseConnection::ptr& conn,const TopicRequest::ptr& msg,Rcode rcode)
        {
            auto msg_rsp = MessageFactory::Createmsg<TopicResponse>();
            msg_rsp->Setid(msg->Getid());
            msg_rsp->SetMtype(Mtype::TOPIC_RES);
            msg_rsp->SetRcode(rcode);
            conn->Send(msg_rsp);
        }
        void TopicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            std::unique_lock<std::mutex> _mutex;
            std::string topic_name = msg->GetKey();
            Topic::ptr topic = std::make_shared<Topic>(topic_name);
            _hashtopics.insert(std::make_pair(topic_name, topic));
        }
        void TopicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            std::string topic_name = msg->GetKey();
            std::unordered_set<Subscriber::ptr> subscribers;
            // 1.在删除主题前，先找到它里面的subscribers
            {
                std::unique_lock<std::mutex> _mutex;
                auto it = _hashtopics.find(topic_name);
                if (it == _hashtopics.end())
                    return;
                subscribers = it->second->_subscribers;
                // 删除主题
                _hashtopics.erase(topic_name);
            }

            
            // 2.删除此主题对应的subscriber
            for (auto &subscriber : subscribers)
            {
                subscriber->RemoveTopic(topic_name);
            }
        }
        bool TopicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 订阅主题
            // 1.先找到对应的主题，没有则创建；找到订阅者，添加此主题
            std::string topic_name = msg->GetKey();
            Topic::ptr topic;
            Subscriber::ptr subs;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _hashtopics.find(topic_name);
                // 没有对应的主题，就返回false
                if (it == _hashtopics.end())
                    return false;
                topic = it->second;

                auto subs_it = _hashconns.find(conn);
                if (subs_it != _hashconns.end())
                    subs = subs_it->second;
                // 没找着，就构造一个订阅者
                else
                {
                    subs = std::make_shared<Subscriber>(conn);
                    _hashconns.insert(std::make_pair(conn, subs));
                }
            }
            // 2.topic添加新订阅者，subs添加新主题
            topic->AddSubscriber(subs);
            subs->AddTopic(topic_name);
            return true;
        }
        void TopicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 取消主题订阅
            // 1.先找到对应的主题删除该订阅者；找到对应的订阅者取关，没有则直接返回
            std::string topic_name = msg->GetKey();
            Topic::ptr topic;
            Subscriber::ptr subs;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _hashtopics.find(topic_name);
                if (it != _hashtopics.end())
                    topic = it->second;

                auto subs_it = _hashconns.find(conn);
                if (subs_it != _hashconns.end())
                    subs = subs_it->second;
            }
            // 2.topic删除对应的订阅者，subs删除对应的主题
            if (subs)
                subs->RemoveTopic(topic_name);
            if (topic && subs)
                topic->RemoveSubscriber(subs);
            
        }
        bool TopicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            // 主题消息发布
            // 1.找到对应的主题，它的内部知道哪些连接订阅了它，直接调用内部的publish函数即可完成 给订阅了这个主题的主机都发送消息
            std::string topic_name = msg->GetKey();
            Topic::ptr topic;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _hashtopics.find(topic_name);
                // 没有对应的主题，就返回false
                if (it == _hashtopics.end())
                    return false;
                topic = it->second;
            }
            topic->PushMessage(msg);
            return true;
        }
        struct Subscriber
        {
            using ptr = std::shared_ptr<Subscriber>;
            std::mutex _mutex;
            BaseConnection::ptr _conn;
            std::unordered_set<std::string> _topics; // 一个订阅者可能订阅多个主题
            Subscriber(const BaseConnection::ptr &conn)
                : _conn(conn)
            {
            }
            // 新订阅主题时调用
            void AddTopic(const std::string &topic)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topics.insert(topic);
            }
            // 主题被删除 或者 取消订阅时调用
            void RemoveTopic(const std::string &topic)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topics.erase(topic);
            }
        };
        struct Topic
        {
            using ptr = std::shared_ptr<Topic>;
            std::mutex _mutex;
            std::string _topic;
            std::unordered_set<Subscriber::ptr> _subscribers; // 这些订阅者都已经订阅了这个主题
            Topic(const std::string &topic)
                : _topic(topic)
            {
            }
            // 新增订阅者时调用
            void AddSubscriber(const Subscriber::ptr &sub)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _subscribers.insert(sub);
            }
            // 订阅者取消订阅此主题 或 订阅者下线 时调用
            void RemoveSubscriber(const Subscriber::ptr &sub)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _subscribers.erase(sub);
            }
            // 收到消息发布请求时调用
            void PushMessage(const BaseMessage::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                for (auto &subscriber : _subscribers)
                {
                    subscriber->_conn->Send(msg);
                }
            }
        };

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Topic::ptr> _hashtopics;             // 一个主题，和Topic内部管理的订阅者连接之间的关联
        std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _hashconns; // 一个连接 和 一个订阅者之间的关联
    };
}